package com.aliyun.flashsale.server.consumer;

import java.util.Collections;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.StaticSessionCredentialsProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
 * 订单创建成功的消息消费
 *
 * 
 * @date 2024/12/05
 */
@Slf4j
@Component
public class OrderSuccessMessageConsumer {

    /**
     * 用户名
     */
    @Value("${rocketmq.username}")
    private String username;
    /**
     * 密码
     */
    @Value("${rocketmq.password}")
    private String password;
    /**
     * 端点
     */
    @Value("${rocketmq.endpoint}")
    private String endpoint;
    /**
     * 主题
     */
    @Value("${rocketmq.order-success-topic}")
    private String topic;
    /**
     * Consumer Group
     */
    @Value("${rocketmq.order-success-consumer-group}")
    private String consumerGroup;
    /**
     * 消费者
     */
    private PushConsumer pushConsumer;

    /**
     * 初始化
     */
    @PostConstruct
    public void init() throws ClientException {
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder configBuilder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
        configBuilder.setCredentialProvider(new StaticSessionCredentialsProvider(username, password));
        ClientConfiguration configuration = configBuilder.build();
        FilterExpression filterExpression = new FilterExpression("*", FilterExpressionType.TAG);
        pushConsumer = provider.newPushConsumerBuilder().setClientConfiguration(configuration).setConsumerGroup(
                consumerGroup).setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
            .setMessageListener(messageView -> {
                // 处理消息并返回消费结果。
                log.info("OrderSuccessMessageConsumer receive message: {}", messageView);
                return ConsumeResult.SUCCESS;
            }).build();
        log.info("OrderSuccessMessageConsumer started");
    }

    /**
     * Destory
     *
     * @throws Exception 例外
     */
    @PreDestroy
    private void destory() throws Exception {
        if (pushConsumer == null) {
            return;
        }
        pushConsumer.close();
    }
}
